package org.bing.zion.core;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;

import org.bing.zion.helper.ChannelBuilder;


public class Connector extends AbstractChannelService {
	private ProtocolFilterChain protocolChain = new ProtocolFilterChain();
	private MessageHandlerChain handlerChain = new MessageHandlerChain();
	protected SocketChannel channel;
	protected Session session;
	protected String ip;
	protected int port;

	protected static volatile int threadNo = 0;

	public Connector(String ip, int port) {
		super(1);
		protocolChain.setMessageDispatcher(new SimpleDispatcher());
		this.ip = ip;
		this.port = port;
	}

	public void connect() throws IOException {
		channel = ChannelBuilder.newSocketChannel(ip, port);
		session = new Session(this, channel, protocolChain, handlerChain);
		channel.register(selector, SelectionKey.OP_CONNECT, session);
		selector.select(DEFAULT_SELECT_TIMEOUT);
		super.read(session);
	}

	protected void doConnect(SelectionKey key) throws IOException {
		channel.register(selector, SelectionKey.OP_READ, session);
	}

	public void addProtocolFilter(ProtocolFilter filter) {
		protocolChain.addFilter(filter);
	}

	public void addMessageHandler(MessageHandler handler) {
		handlerChain.addHandler(handler);
	}

	public void write(Object msg) {
		super.doWrite(session, msg);
	}

	public void writeHeader(String header){
		session.flush(ByteBuffer.wrap(header.getBytes()));
	}
	
	public void close() {
		super.close(channel);
		try {
			if (selector != null) {
				selector.close();
			}
		} catch (IOException e) {
			logger.error("Fail to close " + selector);
		}
	}

	protected void setThreadName() {
		Thread.currentThread().setName("seda-connetor-" + threadNo++);
	}

	@Override
	public void run() {
		setThreadName();
		super.run();
	}
}
